package com.jiayou.server

import cn.hutool.core.date.chinese.SolarTerms
import com.jiayou.*
import com.jiayou.emnu.FrameType
import com.jiayou.emnu.User
import com.jiayou.session.DefaultServerSocketSession
import com.jiayou.session.ServerSocketSession
import com.jiayou.tool.*
import io.ktor.network.selector.*
import io.ktor.network.sockets.*
import io.ktor.util.*
import io.ktor.util.collections.*
import io.ktor.utils.io.*
import kotlinx.coroutines.*
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import java.io.Serializable
import java.net.InetSocketAddress
import java.util.*
import java.util.stream.Collectors
import kotlin.concurrent.thread

/**
@author lishuang
@data   2021-04-23  16:40:20
@info   服务端session服务
 *  T  ->  实际传递数据类型
 *  T1 ->  认证身份类型
 *  T2 ->  认证凭证类型
 */
class ServerSocketServer<T : Serializable, T1 : Serializable, T2 : Serializable>(
    private var serverSocketSession: ServerSocketSession<T, T1, T2>,
    private val port: Int = 9000
) {
    companion object {
        val logger: Logger = LoggerFactory.getLogger(ServerSocketServer::class.java)
    }

    private var server: ServerSocket = aSocket(ActorSelectorManager(Dispatchers.IO))
        .tcp()
        .bind(InetSocketAddress(port))
    private val service = thread(start = false, name = "ktor-server") {
        runBlocking {
            while (true) {
                var authPackPackage: AuthenticatePackage<T, T1, T2>? = null
                val socket = server.accept()
                try {
                    val openReadChannel = socket.openReadChannel()
                    //开启后台协程在指定时间认证
                    GlobalScope.launch {
                        openReadChannel.awaitContent()
                        val availableLength = openReadChannel.readInt()
                        val byteArray = ByteArray(availableLength)
                        openReadChannel.readFully(byteArray, 0, availableLength)
                        val convertResult = byteArrayToObject<T, T1, T2>(byteArray)
                        if (convertResult != null) {
                            if (convertResult.getFrameType() == FrameType.Auth) {
                                val frame = convertResult.getFrame()
                                if (frame is AuthenticatePackage<T, T1, T2>) {
                                    authPackPackage = frame
                                }
                            }
                        }
                    }
                    delay(serverSocketSession.authenticateTime.toLong())
                    //指定时间内没有认证
                    if ((authPackPackage != null)) {
                        if (!onOpen(socket, authPackPackage, openReadChannel))
                            continue
                    } else {
                        continue
                    }
                } catch (e: Exception) {
                    onError(authPackPackage!!.username(), socket)
                    continue
                }


                launch {
                    try {
                        while (true) {
                            if (!onWaitMessage(authPackPackage!!.username(), socket)) break
                        }
                    } catch (e: Throwable) {
                        onError(authPackPackage!!.username(), socket)
                    } finally {
                        ktor.removeUser(socket)
                        DefaultServerSocketSession.logger.error("User   exception connection！")
                    }
                }
                launch {
                    try {
                        onWaitClosed(authPackPackage!!.username(), socket)
                    } catch (e: Throwable) {
                        onError(authPackPackage!!.username(), socket)
                    } finally {
                        ktor.removeUser(socket)
                        DefaultServerSocketSession.logger.error("User   exception connection！")
                    }
                }
            }
        }
    }

    @Synchronized
    fun start() {
        if (!service.isAlive) {
            service.start()
            logger.info("Ktor initialized with port(s): $port (socket)")
        }
    }

    @Synchronized
    fun stop() {
        if (service.isAlive) {
            try {
                service.interrupt()
            } catch (e: InterruptedException) {
                logger.error("Ktor  is cancled")
            } catch (e: Exception) {
                e.printStackTrace()
            }
        }
    }

    @Synchronized
    fun getReaderChannel(): Map<Socket, ByteReadChannel> =
        ktor.stream()
            .collect(Collectors.toMap({ (it[User.SOCKET] ?: error("")).toSocket() }, { (it[User.READ_CHANNEL] ?: error("")).toReadChannel() }, { it1, _ -> it1 }))

    @Synchronized
    fun getWriterChannel(): MutableMap<Socket, ByteWriteChannel> =
        ktor.stream()
            .collect(Collectors.toMap({ (it[User.SOCKET] ?: error("")).toSocket() }, { (it[User.WRITE_CHANNEL] ?: error("")).toWriteChannel() }, { it1, _ -> it1 }))


    @Synchronized
    fun getSocketSession(): MutableSet<Socket> =
        ktor.stream().map { (it[User.SOCKET] ?: error("")).toSocket() }.collect(Collectors.toSet())


    @Synchronized
    fun getSocketTotal(): Int = ktor.size

    private suspend fun onOpen(
        socket: Socket,
        authPackPackage: AuthenticatePackage<T, T1, T2>?,
        readChannel: ByteReadChannel
    ): Boolean =
        serverSocketSession.onOpen(socket, authPackPackage, readChannel)

    private suspend fun onWaitClosed(
        username: T1,
        socket: Socket
    ) = serverSocketSession.onWaitClosed(username, socket)

    private suspend fun onError(username: T1, socket: Socket) = serverSocketSession.onError(username, socket)

    private suspend fun onWaitMessage(
        username: T1,
        socket: Socket
    ) = serverSocketSession.onWaitMessage(username, socket)


    //全局发送消息
    fun sendMessage(broadcastPackage: BroadcastPackage<T, T1, T2>) {
        val protocolFrame = objectToByteArray(DefaultProtocolImpl(FrameType.BROADCAST, broadcastPackage))
        GlobalScope.launch {
            ktor.forEach {
                (it[User.WRITE_CHANNEL] ?: error("")).toWriteChannel().writeInt(protocolFrame.size)
                (it[User.WRITE_CHANNEL] ?: error("")).toWriteChannel().writeFully(protocolFrame)
            }
        }
    }

    //单独发送消息
    fun sendMessage(singlePackage: SinglePackage<T, T1>) {
        val protocolFrame = objectToByteArray(DefaultProtocolImpl(FrameType.SYSTEM, singlePackage))
        GlobalScope.launch {
            val target = ktor.getSocket(singlePackage.to)
            if (target != null) {
                ktor.forEach {
                    if (target == it) {
                        (it[User.WRITE_CHANNEL] ?: error("")).toWriteChannel().writeInt(protocolFrame.size)
                        (it[User.WRITE_CHANNEL] ?: error("")).toWriteChannel().writeFully(protocolFrame)
                        return@forEach
                    }
                }
            }
        }
    }

    //根据条件发送消息
    fun sendMessage(singlPackage: SinglePackage<T, T1>, condition: (username: String) -> Boolean) {
        val protocolFrame = objectToByteArray(DefaultProtocolImpl(FrameType.SYSTEM, singlPackage))
        GlobalScope.launch {
            val target = ktor.getSocket(singlPackage.to)
            if (target != null) {
                ktor.forEach {
                    if (condition.invoke((it[User.USERNAME] ?: error("")).toUsername())) {
                        (it[User.WRITE_CHANNEL] ?: error("")).toWriteChannel().writeInt(protocolFrame.size)
                        (it[User.WRITE_CHANNEL] ?: error("")).toWriteChannel().writeFully(protocolFrame)
                        return@forEach
                    }
                }
            }
        }
    }

    fun <T, T1, T2> getSocketSession() = serverSocketSession


}
